package com.boarsoft.message.core;

import java.io.IOException;
import java.io.ObjectOutput;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.common.Util;
import com.boarsoft.message.MessageConfig;
import com.boarsoft.message.bean.Message;

public class Messenger implements ObjectOutput {
	private static final Logger log = LoggerFactory.getLogger(Messenger.class);

	/** Messenger通过localMsgSvc（localMsgSvcImpl1的实例）发送消息 */
	protected static MessageService messageService;

	public void setMessageService(MessageService service) {
		Messenger.messageService = service;
	}

	/**
	 * 消息生产者（客户端）发送消息<br>
	 * 如果希望先将消息持久化，应在调用此方法前，实现并调用 MessageHelper.prepare 创建消息并持久化
	 * 
	 * @param m
	 * @throws Exception
	 */
	public static void send(Message m) throws Exception {
		log.debug("Send message {}", m);
		// 对消息进行检查
		if (m.isNeedAcknowledge()) {
			// 如果此消息需要回调通知，则必须设置发件人
			if (Util.strIsEmpty(m.getFrom())) {
				throw new IllegalArgumentException("Message.from is required since it need acknowledge");
			}
			// 发件人必须存在？
			// try {
			// messageContext.getConsumer(m.getFrom());
			// } catch (NoSuchBeanDefinitionException e) {
			// throw new IllegalArgumentException(String.format("From %s is
			// invalid", m.getFrom()));
			// }
		}
		// 发送者IP
		m.setFromAddr(MessageConfig.getAddr());
		messageService.put(m);
	}

	@SuppressWarnings("unchecked")
	@Override
	public void writeObject(Object obj) throws IOException {
		if (obj != null && obj instanceof Map) {
			try {
				Message m = this.map2message((Map<String, Object>) obj);
				Messenger.send(m);
			} catch (Exception e) {
				throw new IOException(e);
			}
			return;
		}
		throw new IllegalArgumentException("Message Map only");
	}

	@SuppressWarnings("unchecked")
	protected Message map2message(Map<String, Object> map) {
		Message m = new Message();
		Object ro = map.get("recipients");
		if (ro instanceof List) {
			m.addRecipients((List<String>) ro);
		} else if (ro instanceof Map) {
			m.addRecipients((Map<String, Boolean>) ro);
		} else if (ro == null) {
			// Nothing to to
		} else {
			throw new IllegalArgumentException(//
					"Only List<String> and Map<String, Boolean> be allowed for key 'recipients'");
		}
		// topic（collection）: ebbc_flow_logs
		m.setTarget((String) map.get("target"));
		// 相对于topic的ID（flowId）: flow001
		m.setKey((String) map.get("key"));
		// 附加 flow 的调用链信息
		m.setAttachment(map.get("attachment"));
		// 如果有指定消息创建时间（比如取自日志的时间）
		m.setTime((Long) map.get("time"));
		// 消息内容（日志内容）
		m.setContent((String) map.get("content"));
		// 填入消息来源（发件人）
		m.setFromAddr(MessageConfig.getAddr());
		return m;
	}

	@Override
	public void writeBoolean(boolean v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeByte(int v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeShort(int v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeChar(int v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeInt(int v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeLong(long v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeFloat(float v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeDouble(double v) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeBytes(String s) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeChars(String s) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void writeUTF(String s) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void write(int b) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void write(byte[] b) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void write(byte[] b, int off, int len) throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void flush() throws IOException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void close() throws IOException {
		// throw new UnsupportedOperationException();
	}
}
